//package shortlink.config;
//
//import lombok.RequiredArgsConstructor;
//import org.springframework.context.annotation.Bean;
//import org.springframework.context.annotation.Configuration;
//import org.springframework.data.redis.connection.RedisConnectionFactory;
//import org.springframework.data.redis.connection.stream.Consumer;
//import org.springframework.data.redis.connection.stream.MapRecord;
//import org.springframework.data.redis.connection.stream.ReadOffset;
//import org.springframework.data.redis.connection.stream.StreamOffset;
//import org.springframework.data.redis.stream.StreamMessageListenerContainer;
//import shortlink.mq.consumer.ShortLinkStatsSaveConsumer;
//import org.springframework.data.redis.stream.Subscription;
//import java.time.Duration;
//import java.util.concurrent.ExecutorService;
//import java.util.concurrent.SynchronousQueue;
//import java.util.concurrent.ThreadPoolExecutor;
//import java.util.concurrent.TimeUnit;
//import java.util.concurrent.atomic.AtomicInteger;
//import static shortlink.common.constant.RedisKeyConstant.SHORT_LINK_STATS_STREAM_GROUP_KEY;
//import static shortlink.common.constant.RedisKeyConstant.SHORT_LINK_STATS_STREAM_TOPIC_KEY;
//
//@Configuration
//@RequiredArgsConstructor
//public class RedisStreamConfiguration {
//
//    private final RedisConnectionFactory redisConnectionFactory;
//    private final ShortLinkStatsSaveConsumer shortLinkStatsSaveConsumer;
//
//    @Bean
//    public ExecutorService asyncStreamConsumer() {
//        AtomicInteger index = new AtomicInteger();
//        return new ThreadPoolExecutor(1,
//                1,
//                60,
//                TimeUnit.SECONDS,
//                new SynchronousQueue<>(),
//                runnable -> {
//                    Thread thread = new Thread(runnable);
//                    thread.setName("stream_consumer_short-link_stats_" + index.incrementAndGet());
//                    thread.setDaemon(true);
//                    return thread;
//                },
//                new ThreadPoolExecutor.DiscardOldestPolicy()
//        );
//    }
//
//    @Bean
//    public Subscription shortLinkStatsSaveConsumerSubscription(ExecutorService asyncStreamConsumer) {
//        StreamMessageListenerContainer.StreamMessageListenerContainerOptions<String, MapRecord<String, String, String>> options =
//                StreamMessageListenerContainer.StreamMessageListenerContainerOptions
//                        .builder()
//                        // 一次最多获取多少条消息
//                        .batchSize(10)
//                        // 执行从 Stream 拉取到消息的任务流程
//                        .executor(asyncStreamConsumer)
//                        // 如果没有拉取到消息，需要阻塞的时间。不能大于 ${spring.data.redis.timeout}，否则会超时
//                        .pollTimeout(Duration.ofSeconds(1))
//                        .build();
//        StreamMessageListenerContainer.StreamReadRequest<String> streamReadRequest =
//                StreamMessageListenerContainer.StreamReadRequest.builder(StreamOffset.create(SHORT_LINK_STATS_STREAM_TOPIC_KEY, ReadOffset.lastConsumed()))
//                        .cancelOnError(throwable -> false)
//                        .consumer(Consumer.from(SHORT_LINK_STATS_STREAM_GROUP_KEY, "stats-consumer"))
//                        .autoAcknowledge(true)
//                        .build();
//        StreamMessageListenerContainer<String, MapRecord<String, String, String>> listenerContainer = StreamMessageListenerContainer.create(redisConnectionFactory, options);
//        Subscription subscription = listenerContainer.register(streamReadRequest, shortLinkStatsSaveConsumer);
//
//        listenerContainer.start();
//        return subscription;
//    }
//}
